package org.lemon.transport.context.service.queue.discovery;

import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import lombok.extern.slf4j.Slf4j;
import org.lemon.transport.context.data.tenant.TenantId;
import org.lemon.transport.context.service.discovery.ServiceInfoProvider;
import org.lemon.transport.context.service.queue.ServiceType;
import org.lemon.transport.context.service.queue.TopicPartitionInfo;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Slf4j
@Service
public class HashPartitionService {

    private final String hashFunctionName = "murmur3_128";

    private ConcurrentMap<QueueKey, List<Integer>> myPartitions = new ConcurrentHashMap<>();

    private final ConcurrentMap<QueueKey, String> partitionTopicsMap = new ConcurrentHashMap<>();
    private final ConcurrentMap<QueueKey, Integer> partitionSizesMap = new ConcurrentHashMap<>();

    private final ConcurrentMap<TenantId, TenantRoutingInfo> tenantRoutingInfoMap = new ConcurrentHashMap<>();

    private HashFunction hashFunction;

    private final ApplicationEventPublisher applicationEventPublisher;
    private final ServiceInfoProvider serviceInfoProvider;
    private final TenantRoutingInfoService tenantRoutingInfoService;
    private final QueueRoutingInfoService queueRoutingInfoService;

    public HashPartitionService(ApplicationEventPublisher applicationEventPublisher,
                                ServiceInfoProvider serviceInfoProvider,
                                TenantRoutingInfoService tenantRoutingInfoService,
                                QueueRoutingInfoService queueRoutingInfoService) {
        this.applicationEventPublisher = applicationEventPublisher;
        this.serviceInfoProvider = serviceInfoProvider;
        this.tenantRoutingInfoService = tenantRoutingInfoService;
        this.queueRoutingInfoService = queueRoutingInfoService;
    }

    @PostConstruct
    public void init() {
        this.hashFunction = forName(hashFunctionName);
    }

    public TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, Long entityId) {

        TenantId isolatedOrSystemTenantId = getIsolatedOrSystemTenantId(serviceType, tenantId);
        QueueKey queueKey = new QueueKey(serviceType, queueName, isolatedOrSystemTenantId);
        if (!partitionSizesMap.containsKey(queueKey)) {
            queueKey = new QueueKey(serviceType, isolatedOrSystemTenantId);
        }
        return resolve(queueKey, entityId);
    }

    public TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, Long entityId) {
        return resolve(serviceType, null, tenantId, entityId);
    }

    private TopicPartitionInfo resolve(QueueKey queueKey, Long entityId) {

        int hash = hashFunction.newHasher()
                .putLong(entityId)
                .hash()
                .asInt();

        Integer partitionSize = partitionSizesMap.get(queueKey);
        int partition = Math.abs(hash % partitionSize);

        return buildTopicPartitionInfo(queueKey, partition);
    }

    private TenantId getIsolatedOrSystemTenantId(ServiceType serviceType, TenantId tenantId) {
        return isIsolated(serviceType, tenantId) ? tenantId : TenantId.SYS_TENANT_ID;
    }

    private boolean isIsolated(ServiceType serviceType, TenantId tenantId) {
        if (TenantId.SYS_TENANT_ID.equals(tenantId)) {
            return false;
        }
        TenantRoutingInfo routingInfo = tenantRoutingInfoMap.get(tenantId);
        if (routingInfo == null) {
            synchronized (tenantRoutingInfoMap) {
                routingInfo = tenantRoutingInfoMap.get(tenantId);
                if (routingInfo == null) {
                    routingInfo = tenantRoutingInfoService.getRoutingInfo(tenantId);
                    tenantRoutingInfoMap.put(tenantId, routingInfo);
                }
            }
        }
        if (routingInfo == null) {
            throw new RuntimeException("Tenant not found!");
        }
        switch (serviceType) {
            case RULE_ENGINE:
                return routingInfo.isIsolatedRuleEngine();
            default:
                return false;
        }
    }

    private TopicPartitionInfo buildTopicPartitionInfo(QueueKey queueKey, int partition) {

        List<Integer> partitions = myPartitions.get(queueKey);
        boolean myPartition = false;
        if (partitions != null) {
            myPartition = partitions.contains(partition);
        }
        return new TopicPartitionInfo(partitionTopicsMap.get(queueKey),
                queueKey.getTenantId(),
                partition,
                myPartition);

    }

    public static HashFunction forName(String name) {
        switch (name) {
            case "murmur3_32":
                return Hashing.murmur3_32();
            case "murmur3_128":
                return Hashing.murmur3_128();
            case "sha256":
                return Hashing.sha256();
            default:
                throw new IllegalArgumentException("Can't find hash function with name " + name);
        }
    }
}
